package com.we.flink.dcapiuserlabel;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.we.flink.utils.WeKafkaPropertyReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

public class AdmTelsaleUserLabel {
    public static final String INPUT_KEY_BY = "connect_id";
    public static final String OUTPUT_KEY_BY = "uid";
    public static final String RELEASEPROP = "dcapiuserLabel/telsale/telsalelabel_prod.properties";
    public static final String USERLABELPROP = "dcapiuserLabel/telsale/telsalelabel.properties";
    public static Logger LOG = LoggerFactory.getLogger(AdmTelsaleUserLabel.class);

    public static void main(String[] args) throws IOException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        try {
            WeKafkaPropertyReader paramReader = WeKafkaPropertyReader.init(RELEASEPROP);
            /** RocksDB */
            env.setStateBackend(new RocksDBStateBackend(paramReader.getRocksDBBackendUrl()));
            /** checkpoint configure */
            CheckpointConfig ckConf = env.getCheckpointConfig();
            ckConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            ckConf.setCheckpointInterval(10 * 60 * 1000); // ms
            ckConf.setCheckpointTimeout(60 * 60 * 1000);
            ckConf.setMaxConcurrentCheckpoints(1);
            ckConf.setMinPauseBetweenCheckpoints(500);
            ckConf.enableExternalizedCheckpoints(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

            ckConf.enableUnalignedCheckpoints();

            /** Consumer Kafka */
            String srcKafkaTopic = paramReader.getTmpKfkTopic();
            String srcKafkaBootStrapServer = paramReader.getTmpKfkBootStrapServer();
            String srckfkGrupId = paramReader.getTmpKfkGroupId();
            String srckfkOffset = paramReader.getTmpKfkOffset();

            Properties consumProp = new Properties();
            consumProp.setProperty("bootstrap.servers", srcKafkaBootStrapServer);
            consumProp.setProperty("group.id", srckfkGrupId);
            consumProp.setProperty("auto.offset.reset", srckfkOffset);
            consumProp.setProperty(
                    "key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            consumProp.setProperty(
                    "value.deserializer",
                    "org.apache.kafka.common.serialization.StringDeserializer");

            //            FlinkKafkaConsumer<String> kfkSource = new
            // FlinkKafkaConsumer<String>(srcKafkaTopic, new SimpleStringSchema(), consumProp);
            //            DataStreamSource<String> input = env.addSource(kfkSource);

            //            KafkaSource<String> source = KafkaSource.<String>builder()
            //                    .setBootstrapServers(brokers)
            //                    .setTopics("input-topic")
            //                    .setGroupId("my-group")
            //                    .setStartingOffsets(OffsetsInitializer.earliest())
            //                    .setValueOnlyDeserializer(new SimpleStringSchema())
            //                    .build();
            KafkaSource<String> kfkSource =
                    KafkaSource.<String>builder()
                            .setTopics(srcKafkaTopic)
                            .setProperties(consumProp)
                            .setValueOnlyDeserializer(new SimpleStringSchema())
                            .build();

            DataStreamSource<String> input =
                    env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "kfkSource");

            SingleOutputStreamOperator<String> out =
                    input.keyBy(
                                    new KeySelector<String, Integer>() {
                                        @Override
                                        public Integer getKey(String value) throws Exception {
                                            return JSONObject.parseObject(value)
                                                    .getIntValue(INPUT_KEY_BY);
                                        }
                                    })
                            .map(new TelsaleLabelRichMapFunc())
                            .filter(
                                    new FilterFunction<String>() {
                                        @Override
                                        public boolean filter(String value) throws Exception {
                                            JSONObject jsonObject = JSONObject.parseObject(value);
                                            if (jsonObject.containsKey(OUTPUT_KEY_BY)) {
                                                return (jsonObject.get(OUTPUT_KEY_BY) != null);
                                            } else {
                                                return false;
                                            }
                                        }
                                    })
                            .keyBy(
                                    new KeySelector<String, Long>() {
                                        @Override
                                        public Long getKey(String value) throws Exception {
                                            return JSONObject.parseObject(value)
                                                    .getLong(OUTPUT_KEY_BY);
                                        }
                                    })
                            .flatMap(new TelsaleLabelRichFlatMapFunc());

            //                out.print();
            /** sink to Kafka // */
            String sinkkfkTopic = paramReader.getKfkTopic();
            Properties sinkProp = new Properties();
            sinkProp.setProperty("bootstrap.servers", paramReader.getKfkBootStrapServer());
            sinkProp.setProperty("acks", "all");
            int sinkkfkPartitions = paramReader.getKfkPartitions();

            FlinkKafkaProducer<String> userLabelAllFlinkKafkaProducer =
                    new FlinkKafkaProducer<String>(
                            sinkkfkTopic,
                            new WeKafkaKeyedSerializationSchema(),
                            sinkProp,
                            java.util.Optional.of(new WeKafkaCustomPartitioner()));

            out.addSink(userLabelAllFlinkKafkaProducer).setParallelism(sinkkfkPartitions);

            env.execute(AdmTelsaleUserLabel.class.toString());
        } catch (Exception e) {
            System.out.println("Exception: " + e.getMessage());
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static class TelsaleLabelRichMapFunc extends RichMapFunction<String, String> {
        ValueState<JSONObject> userLabelVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            userLabelVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "userLabelVState", JSONObject.class));
        }

        @Override
        public String map(String value) throws Exception {

            JSONObject tmp = userLabelVState.value();
            JSONObject curState = (tmp == null ? initJsonObjState(USERLABELPROP) : tmp);
            JSONObject input = JSON.parseObject(value);
            for (Map.Entry<String, Object> inputEle : input.entrySet()) {
                String key = inputEle.getKey();
                Object val = inputEle.getValue();
                if (curState.containsKey(key)) {
                    curState.put(key, val);
                } else {
                    LOG.error("Properties not contains key: " + key + " value: " + val);
                }
            }
            userLabelVState.update(curState);
            /** set account_user_key to null to avoid output account_user_key */
            //            System.out.println("##Dbg: RichMap" + curState.toString());
            LOG.debug("##Dbg: RichMap" + curState.toString());
            return curState.toString();
        }
    }

    private static class TelsaleLabelRichFlatMapFunc extends RichFlatMapFunction<String, String> {
        ValueState<JSONObject> userLabelVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            userLabelVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<JSONObject>(
                                            "TelsaleLabelVState", JSONObject.class));
        }

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            JSONObject tmp = userLabelVState.value();
            JSONObject curState = (tmp == null ? initJsonObjState(USERLABELPROP) : tmp);
            System.out.println("input: " + value);
            JSONObject input = JSON.parseObject(value);
            Long userKeyVal = input.getLong(OUTPUT_KEY_BY);
            //            System.out.println("output_key: " + userKeyVal);
            for (Map.Entry<String, Object> inputEle : input.entrySet()) {
                String key = inputEle.getKey();
                Object val = inputEle.getValue();
                //                System.out.println("key: " + key + "  val: " + val.toString());
                if (!key.equals(INPUT_KEY_BY) && !key.equals(OUTPUT_KEY_BY)) {
                    Object stateVal = curState.get(key);
                    //                    System.out.println("curstate value: " + stateVal);
                    if (stateVal != null && val != null) {
                        if (!stateVal.equals(val)) {
                            curState.put(key, val);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                            jsonObject.put(key, val);
                            //                            System.out.println("out1: " +
                            // jsonObject.toString());
                            out.collect(jsonObject.toString());
                        }
                    } else {
                        if (stateVal != val) {
                            curState.put(key, val);
                            JSONObject jsonObject = new JSONObject();
                            jsonObject.put(OUTPUT_KEY_BY, userKeyVal);
                            jsonObject.put(key, val);
                            //                            System.out.println("out2: " +
                            // jsonObject.toString());
                            out.collect(jsonObject.toString());
                        }
                    }
                }
            }
            userLabelVState.update(curState);
        }
    }

    private static JSONObject initJsonObjState(String propFile) throws IOException {
        JSONObject res = new JSONObject();
        InputStream resourceAsStream =
                AdmTelsaleUserLabel.class.getClassLoader().getResourceAsStream(propFile);
        ParameterTool reader = ParameterTool.fromPropertiesFile(resourceAsStream);
        Set<Object> propSets = reader.getProperties().keySet();

        for (Object key : propSets) {
            String typeValue = reader.get((String) key);
            String[] splits = typeValue.split(":");
            if (splits.length == 2) {
                String type = splits[0];
                String value = splits[1];
                jsonObjectSetValue(res, type, (String) key, value);
            } else {
                LOG.error("split Error: " + splits.length);
            }
        }
        return res;
    }

    private static void jsonObjectSetValue(
            JSONObject jsonObj, String type, String key, String value) {
        if (!value.equals("null")) {
            switch (type) {
                case "Long":
                    jsonObj.put(key, Long.parseLong(value));
                    break;
                case "String":
                    jsonObj.put(key, value);
                    break;
                case "Float":
                    jsonObj.put(key, Float.parseFloat(value));
                    break;
                case "Integer":
                    jsonObj.put(key, Integer.parseInt(value));
                    break;
                default:
                    LOG.error("Not support this type: " + type);
                    break;
            }
        } else {
            jsonObj.put(key, null);
        }
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getLong(OUTPUT_KEY_BY).toString();
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
